package com.shujia.flink.core

import java.util

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo5ListState {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Int)] = ds.map(line => {
      val split: Array[String] = line.split(",")

      val clazz: String = split(4)

      val age: String = split(2)

      (clazz, age.toInt)
    })

    /**
      *
      * 实时统计班级平均年龄
      *
      */

    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)


    val avgDS: DataStream[(String, Double)] = keyByDS.process(new AvgAgeKeyedProcessFunction)

    avgDS.print()


    env.execute()


  }
}

class AvgAgeKeyedProcessFunction extends KeyedProcessFunction[String, (String, Int), (String, Double)] {

  var state: ListState[Int] = _

  override def open(parameters: Configuration): Unit = {

    val context: RuntimeContext = getRuntimeContext

    val listState = new ListStateDescriptor[Int]("avg", classOf[Int])

    /**
      * ListState：为每一个key保存多个值的状态
      *
      */

    state = context.getListState(listState)

  }

  override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Double)]#Context, out: Collector[(String, Double)]): Unit = {


    //往集合状态中保存一个值
    state.add(value._2)


    //获取liststate中所有的值
    val iter: util.Iterator[Int] = state.get().iterator()

    //导入java集合转换成scala集合的隐式转换
    import scala.collection.JavaConversions._

    val list: List[Int] = iter.toList


    //计算平均年龄
    val avgAge: Double = list.sum.toDouble / list.length


    //将数据发送到下游
    out.collect((value._1, avgAge))

  }
}
